package com.ydlclass.core;

import com.ydlclass.NettyBootstrapInitializer;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.compress.CompressorFactory;
import com.ydlclass.discovery.Registry;
import com.ydlclass.enumeration.RequestType;
import com.ydlclass.serialize.SerializerFactory;
import com.ydlclass.transport.message.YrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 心跳探测的核心目的是什么？探活，感知哪些服务器的连接状态是正常的，哪些是不正常的
 *
 */
@Slf4j
public class HeartbeatDetector {

    /**
     * 在客户端通过 YrpcBootstrap.reference(referenceConfig) 引用一个新服务时被触发。
     * 这意味着，一旦客户端对某个服务产生了兴趣（引用了它），
     * 就会启动一个专门的后台线程，开始对该服务的所有提供者进行周期性的心跳探测。
     * @param ServiceName 服务名称
     */
    public static void detectHeartbeat(String ServiceName) {
        // 1、从注册中心拉取服务列表
        Registry registry = YrpcBootstrap.getInstance().getConfiguration().getRegistryConfig().getRegistry();
        List<InetSocketAddress> addresses = registry.lookup(ServiceName,
                YrpcBootstrap.getInstance().getConfiguration().getGroup());

        // 2.遍历列表，为每一个InetSocketAddress对象建立连接，并将连接进行缓存。
        /**
         * 进行缓存的目的是避免重复创建和销毁Netty Channel，提升性能并保证长连接的稳定性
         * 如果不进行缓存的话，每一次心跳检测都会进行TCP的三次握手，会增加时间开销
         * 
         * 在开始探测之前，首先确保与该服务的所有提供者节点都建立了Netty连接，
         * 并将这些连接（Channel）缓存到全局的 YrpcBootstrap.CHANNEL_CACHE 中。这为后续的探测做好了准备。
         */
        for (InetSocketAddress address : addresses) {
            try {
                if (!YrpcBootstrap.CHANNEL_CACHE.containsKey(address)) {
                    Channel channel = NettyBootstrapInitializer
                            .getBootstrap() // 获取单例
                            .connect(address) // 向目标地址发起TCP非阻塞的连接
                            .sync() // 阻塞当前线程，直到连接成功建立或失败
                            .channel(); // 获取连接成功的channel
                    YrpcBootstrap.CHANNEL_CACHE.put(address, channel);// 缓存连接进行复用
                }

            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        // 3、任务，定期发送消息
        /**
         * Timer的缺陷：1.Timer 是单线程的，如果 MyTimerTask 抛出未捕获的异常，整个 Timer 会终止。
         *             2.Timer 的任务是串行执行的，如果某个任务执行时间过长，会影响后续任务的调度。
         * 后续使用线程池解决
         */
        Thread thread = new Thread(() -> new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 2000),
                "yrpc-HeartbeatDetector-thread"); //立即执行一次任务，然后每隔 2000 毫秒（2秒）再执行一次。
        thread.setDaemon(true); // 创建守护线程，该线程不会阻止JVM的退出
        thread.start();
    }

    /**
     * 心跳检测任务
     */
    private static class MyTimerTask extends TimerTask { // 心跳检测属于生命周期、独立线程运行的任务，使用静态内部类避免内存泄漏。
    
        @Override
        public void run() {

            // 将响应时长的map清空
            YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            // 遍历所有的channel，每一个channel都要进行心跳检测
            Map<InetSocketAddress, Channel> cache = YrpcBootstrap.CHANNEL_CACHE;
            for (Map.Entry<InetSocketAddress, Channel> entry : cache.entrySet()) {
                // 定义一个重试的次数
                int tryTimes = 3;
                while (tryTimes > 0) {
                    // 通过心跳检测处理每一个channel
                    Channel channel = entry.getValue();

                    long start = System.currentTimeMillis();
                    // 构建一个心跳请求（不需要带请求体）
                    YrpcRequest yrpcRequest = YrpcRequest.builder()
                            .requestId(YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                            .compressType(CompressorFactory.getCompressor(YrpcBootstrap.getInstance()
                                    .getConfiguration().getCompressType()).getCode())
                            .requestType(RequestType.HEART_BEAT.getId())
                            .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance()
                                    .getConfiguration().getSerializeType()).getCode())
                            .timeStamp(start)
                            .build();

                    // 4、写出报文

                    CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                    // 将 completableFuture 暴露出去
                    YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(), completableFuture);

                    /**
                     * 发送心跳请求
                     */
                    channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                        if (!promise.isSuccess()) {
                            completableFuture.completeExceptionally(promise.cause());
                        }
                    });

                    Long endTime = 0L;
                    try {
                        // 阻塞方法，get()方法如果得不到结果，就会一直阻塞
                        // 我们想不一直阻塞可以添加参数
                        completableFuture.get(1, TimeUnit.SECONDS);
                        endTime = System.currentTimeMillis();
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
                        // 一旦发生问题，需要优先重试
                        tryTimes--;

                        /**
                         * 会执行一次尝试，两次重试，共三次机会，如果三次全部失败，将channel移除缓存

                         */
                        log.error("和地址为【{}】的主机连接发生异常.正在进行第【{}】次重试......",
                                channel.remoteAddress(), 3 - tryTimes); 

                        // 如果重试的机会用尽，就将失效的地址移出服务缓存列表
                        if (tryTimes == 0) {
                            YrpcBootstrap.CHANNEL_CACHE.remove(entry.getKey());
                        }

                        // 尝试等待一段时间后再重试，不要立即重试。
                        try {
                            Thread.sleep(10 * (new Random().nextInt(5)));
                        } catch (InterruptedException ex) {
                            throw new RuntimeException(ex);
                        }

                        continue;
                    }
                    Long time = endTime - start; //计算响应时间

                    // 使用treemap进行缓存
                    YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time, channel);
                    log.debug("和[{}]服务器的响应时间是[{}].", entry.getKey(), time);
                    break; //没有捕获到异常直接执行break
                }
            }

            log.info("-----------------------响应时间的treemap----------------------");
            for (Map.Entry<Long, Channel> entry : YrpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}]--->channelId:[{}]", entry.getKey(), entry.getValue().id());
                }
            }
        }
    }

}
